# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: xguo

import collections
import enum
import functools
import io
import json
import typing
import dateutil.parser

import requests
import tornado.gen
import tornado.httpclient
import recordclass

import coin.flow.subscriber as flow_subscriber
import coin.flow.topic_map as topic_map
import coin.proto.coin_market_enums_pb2 as coin_enum

from coin.base.config import Config
from coin.base.timestamp import get_timestamp
from coin.exchange.base.fill_stats import FillStatsCalculator2
from coin.exchange.base.order_fill_util import FillManager, LastFill
from coin.exchange.base.order_gateway import BaseOrderGatewayInfoMutator
from coin.exchange.base.order_gateway_ready import PositionOrderGatewayReady
from coin.exchange.base.rest.task_scheduler import Scheduler, wrap_async_task
from coin.exchange.okex_futures.kr_rest.currency import OkexCurrency
from coin.exchange.okex_futures.kr_rest.futures_product import OkexFuturesProduct
from coin.exchange.okex_futures_v3.kr_rest.private_client import OkexFuturesPrivateParser
from coin.exchange.okex_futures_v3.ws.private_subscriber import OkexFuturesWsPrivateClient
from coin.proto.coin_order_gateway_pb2 import OrderGatewayConfig, OrderEvent
from coin.strategy.hk_hard_arb.og_util import print_working_orders
from coin.exchange.okex_futures_v3.kr_rest.native_private_client import (
    OkexFuturesNativePrivateClient)
from coin.exchange.base.order_gateway_util import (
    PrivateExchangeMessage,
    convert_to_requests_response,
)
from coin.proto.coin_query_pb2 import (
    CurrencyBalance,
    AccountBalance,
    AccountPosition,
    ProductPosition,
)
from coin.exchange.okex_v3.kr_rest.native_private_client import (
    OkexFuturesOrderSide,
    OkexOrderType,
    OkexFuturesOrderStatus,
    OkexLeverageType,
    OkexOrderProperty,
)
from coin.exchange.base.rest_order_gateway_base import (
    RestOrderGateway,
    exchange_message_handler,
    create_order_internal,
)
from coin.exchange.base.order_gateway import (
    Order,
    OrderType,
    OrderGatewayStatus,
)


def create_okex_order_from_orderinfo(data, timestamp, order_id=None):
  native_symbol = data['instrument_id']
  product = OkexFuturesProduct.FromStrNativeProduct(native_symbol)
  exchange_order_id = str(data['order_id'])
  price_orig = float(data['price'])
  price_avg_fill = float(data['price_avg'])
  qty_orig = float(data['size'])
  qty_accum_fill = float(data['filled_qty'])
  fee = float(data['fee'])
  future_order_side = OkexFuturesOrderSide(int(data['type']))
  order_type = OrderType.LIMIT
  status = OkexFuturesOrderStatus(int(data['status']))
  order_creation_time = None
  if 'timestamp' in data:
    order_creation_time = int(dateutil.parser.parse(data['timestamp']).timestamp() * 1e9)
  fill = FillManager(price_orig=price_orig,
                     qty_orig=qty_orig,
                     qty_last_fill=qty_accum_fill,
                     price_last_fill=price_avg_fill,
                     qty_accum_fill=qty_accum_fill,
                     price_avg_fill=price_avg_fill)
  internal = create_order_internal(
      exchange_order_id=exchange_order_id,
      fill=fill,
      timestamp=timestamp,
      status=status,
      fee=fee,
  )
  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=future_order_side,
                order_type=order_type,
                tag={'order_creation_time': order_creation_time},
                accepted=True,
                cancel_sent=False,
                internal=internal)
  return order


class OkexFuturesFillStatsCalculator(FillStatsCalculator2):
  ProductType = OkexFuturesProduct
  BuyOrderSide = [OkexFuturesOrderSide.BUY_OPEN, OkexFuturesOrderSide.BUY_CLOSE]
  SellOrderSide = [OkexFuturesOrderSide.SELL_OPEN, OkexFuturesOrderSide.SELL_CLOSE]


OkexFuturesPosition = recordclass.recordclass('OkexFuturesPosition', [
    'long',
    'short',
    'long_avg_price',
    'short_avg_price',
])


def create_futures_positions(long=0., short=0., long_avg_price=None, short_avg_price=None):
  return OkexFuturesPosition(long=long,
                             short=short,
                             long_avg_price=long_avg_price,
                             short_avg_price=short_avg_price)


class OkexFuturesOrderGatewayInfo(BaseOrderGatewayInfoMutator):
  CurrencyType = OkexCurrency
  ProductType = OkexFuturesProduct
  BalanceType = float
  PositionType = float

  def __init__(self,
               currencies: typing.List[OkexCurrency],
               products: typing.List[OkexFuturesProduct]):
    super().__init__(currencies, products)
    self._unrealized_pnl = collections.defaultdict(float)
    self._clear_balance()
    self._clear_position()

  def _clear_balance(self):
    for currency in self.currencies:
      balance = CurrencyBalance(currency=currency.currency,
                                currency_native=currency.native_currency,
                                total=0.,
                                available=0.,
                                hold=0.)
      self.set_balance(currency, 0.0, balance)

  def _clear_position(self):
    for product in self.products:
      position = ProductPosition(symbol=product.symbol,
                                 long_position=0.,
                                 short_position=0.,
                                 net_position=0.)
      self.set_position(product, 0.0, position)

  def update_balance(self, msg):
    multiple_queries = msg.get('multiple_queries')
    account_balance = OkexFuturesPrivateParser.parse_cross_margin_balance(msg)
    if not multiple_queries:
      self._clear_balance()
    for balance in account_balance.each_balance:
      currency = OkexCurrency.FromStr(balance.currency)
      self.set_balance(currency, balance.total, balance)

    for native_currency, balance_info in msg['info'].items():
      currency = OkexCurrency.FromStrNativeCurrency(native_currency)
      self._unrealized_pnl[currency.currency] = float(balance_info['unrealized_pnl'])

  def update_position(self, message):
    msg = OkexFuturesPrivateParser.parse_position(message)
    self._clear_position()
    for position in msg.each_position:
      product = OkexFuturesProduct.FromStrProduct(position.symbol)
      self.set_position(product, position.net_position, position)

  def update_ws_position(self, message):
    position = OkexFuturesPrivateParser.parse_ws_product_position(message)
    product = OkexFuturesProduct.FromStrProduct(position.symbol)
    self.set_position(product, position.net_position, position)

  def balance_as_proto(self):
    return AccountBalance(exchange='Okex',
                          market_type='Futures',
                          each_balance=self.get_notNone_balances_internal())

  def position_as_proto(self):
    return AccountPosition(exchange='Okex',
                           market_type='Futures',
                           each_position=self.get_notNone_positions_internal())

  def get_product_position_proto(self, product):
    return self._positions_internal[product.symbol]

  def get_okex_futures_position(self, product: OkexFuturesProduct):
    """return FuturesPosition namedtuple"""
    position = self.get_product_position_proto(product)
    return OkexFuturesPosition(long=position.long_position,
                               short=abs(position.short_position),
                               long_avg_price=position.long_avg_entry_price,
                               short_avg_price=position.short_avg_entry_price)

  # Margin balance includes unrealized pnl, but this is provided by exchange
  # and could be out of date or slow.
  # DO NOT USE THIS IN TRADING, ONLY ACCEPTABLE IN PNL DISPLAY.
  def get_margin_balance(self, currency: str):
    return self._unrealized_pnl[currency] + self.get_balance(currency)


class OkexFuturesOrderGatewayBase(RestOrderGateway):
  _exchange = 'Okex'
  _market_type = 'Futures'

  OGInfoType = OkexFuturesOrderGatewayInfo

  class _MsgType(enum.IntEnum):
    UNKNOWN = 0
    REST_BALANCE = 1
    REST_POSITION = 2
    REST_OPEN_ORDERS = 3
    REST_DONE_ORDERS = 4
    WS_POSITION = 5
    WS_ORDER_UPDATE = 6

  def __init__(self,
               currencies: typing.List[OkexCurrency],
               products: typing.List[OkexFuturesProduct],
               config: Config,
               og_config=None,
               logger=None):
    super().__init__(currencies, products, config, og_config, logger)
    self._og_ready = PositionOrderGatewayReady()
    self._og_ready.set_status(OrderGatewayStatus.INITIALIZING)

    self._private_client = None
    self._private_subscriber = None
    self._scheduler = None

    self._ioloop = None
    self._flow_sub = None
    self._publisher = None
    self._start_timestamp = get_timestamp()

    default_og_config = OrderGatewayConfig(
        account_query_period=5,  # Account query frequency is restrictive.
        user_query_period=1,  # Position query
        orders_snapshot_query_period=1,
        rest_lib='pycurl',
    )
    default_og_config.MergeFrom(self._og_config)
    self._og_config = default_og_config
    self._og_logger.exchange_type = coin_enum.Okex
    self._og_logger.market_type = coin_enum.Futures
    self._fill_stats_calculator = OkexFuturesFillStatsCalculator(products)
    self._og_info_rest = self._og_info  # For v1 compatibility
    self._num_done_order_query = 0
    self._channel_msg_type_map = {
        'futures/order': self._MsgType.WS_ORDER_UPDATE,
        'swap/order': self._MsgType.WS_ORDER_UPDATE,
        'futures/position': self._MsgType.WS_POSITION,
        'swap/position': self._MsgType.WS_POSITION,
    }
    self._logger.info('OG currencies: %s\n',
                      json.dumps(self._og_info.currencies, default=str, indent=2))
    self._logger.info('OG products: %s\n',
                      json.dumps(self._og_info.products, default=str, indent=2))

  def get_okex_futures_position(self, product: OkexFuturesProduct):
    return self._og_info.get_okex_futures_position(product)

  def _prepare_private_subscriber(self):
    raise NotImplementedError()

  def get_fill_stats(self, product):
    return self._fill_stats_calculator.get_fill_stats(product)

  def examine_rest_response_timestamp(self, exchange_message: PrivateExchangeMessage):
    delta_time = exchange_message.received_timestamp - exchange_message.request_timestamp
    if delta_time > 1e9:
      self._logger.warning('Rest response is slow!')

  def submittable(self):
    return self.is_ready()

  @tornado.gen.coroutine
  def _query_account_balance(self):
    # emulate a single call of account balance query
    request_timestamp = get_timestamp()
    aggregated = {'result': True, 'info': {}, 'multiple_queries': True}

    usdt_currency = set()
    for product in self._og_info._products:
      if product.quote == "USDT":
        usdt_currency.add(f"{product.base.native_currency}-{product.quote.native_currency}")

    assert len(usdt_currency) <= 1

    for currency in self._og_info.currencies + list(usdt_currency):
      if isinstance(currency, str):
        native_currency = currency
        balance_name = "USDT"
      else:
        native_currency = currency.native_currency_v3
        balance_name = native_currency
        if currency.currency == 'USD':  # no USD in Okex Futures
          continue
        elif currency.currency == 'USDT':
          continue

      response = yield self._private_client.query_single_account(native_currency)
      response = convert_to_requests_response(response)
      if response.status_code != 200:
        self._logger.error('query %s balance failed!' % native_currency)
        continue
      balance_info = response.json()
      aggregated['info'][balance_name] = balance_info

    response = requests.Response()
    response.url = ('query_account_balance (%s, %s, %s)' %
                    (self._exchange, self._market_type, self._api_version))
    response.status_code = 200
    response.encoding = 'UTF-8'
    response.raw = io.BytesIO(json.dumps(aggregated).encode())
    timestamp = get_timestamp()
    msg = PrivateExchangeMessage(
        msg_type=self._MsgType.REST_BALANCE,
        data=response,
        received_timestamp=timestamp,
        request_params=None,
        request_timestamp=request_timestamp,
    )
    self._publisher.write(timestamp, msg)

  @tornado.gen.coroutine
  def _install_periodic_query(self):
    assert self._scheduler is not None, 'self._scheduler is not set!'
    assert self._publisher is not None, 'self._publisher is not set!'

    # query account balance
    self._scheduler.add_periodic_task(period=self._og_config.account_query_period,
                                      job_func=self._query_account_balance)

    # query positions
    yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
    query_position_task = wrap_async_task(
        self._private_client.query_positions,
        msg_type=self._MsgType.REST_POSITION,
        publisher=self._publisher,
    )
    self._scheduler.add_periodic_task(
        period=self._og_config.user_query_period,
        job_func=query_position_task,
    )

    for product in self._og_info.products:
      instrument_id = product.native_symbol_v3
      # query active orders
      yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
      query_orders_task = wrap_async_task(
          self._private_client.query_order_list,
          msg_type=self._MsgType.REST_OPEN_ORDERS,
          publisher=self._publisher,
          func_params=dict(instrument_id=instrument_id, status=OkexFuturesOrderStatus.OPEN),
      )
      self._scheduler.add_periodic_task(
          period=self._og_config.orders_snapshot_query_period,
          job_func=query_orders_task,
      )

      # query completed orders (filled + cancelled)
      yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
      query_fills_task = wrap_async_task(
          self._private_client.query_order_list,
          msg_type=self._MsgType.REST_DONE_ORDERS,
          publisher=self._publisher,
          func_params=dict(instrument_id=instrument_id, status=OkexFuturesOrderStatus.DONE),
      )
      self._scheduler.add_periodic_task(
          period=self._og_config.fill_query_period,
          job_func=query_fills_task,
      )

    def print_info():
      print_working_orders(self._logger, self)

    # Periodically print working order list.
    self._scheduler.add_periodic_task(period=10, job_func=print_info)
    self._scheduler.start()

  def _handle_error_response(self, response):
    if response.status_code == requests.codes.too_many_requests:  # 429
      self._logger.error('Rate limit hit! status_code = %s, %s, %s',
                         response.status_code,
                         response.url,
                         response.text)
    else:
      self._logger.error('Query error! status_code = %s, %s, %s',
                         response.status_code,
                         response.url,
                         response.text)

  def _async_submit_order(self, order, post_only=None):
    assert order.order_id is not None, order
    assert order.order_type == OrderType.LIMIT, order
    order_property = None
    if post_only:
      order_property = OkexOrderProperty.POST_ONLY
    else:
      order_property = None

    order.internal.leverage = OkexLeverageType(order.internal.leverage)
    formatter = self._order_formatter[order.product]
    params = {
        'instrument_id': order.product.native_symbol_v3,
        'price': formatter.format_price(order.price),
        'size': formatter.format_qty(order.qty),
        'order_side': order.order_side,
        'order_type': OkexOrderType(order.order_type.name.lower()),
        'order_property': order_property,
        'leverage': order.internal.leverage,
    }
    fut = self._private_client.place_order(**params)
    return fut

  def _async_cancel_order(self, order):
    assert order.internal.exchange_order_id is not None, order
    fut = self._private_client.cancel_order(
        order_id=order.internal.exchange_order_id,
        instrument_id=order.product.native_symbol_v3,
    )
    return fut

  def _handle_submit_response(self, order, response):
    """
    {
    "client_oid": "hello",
    "order_id": "1680287676308480",
    "result": true
    }
    """
    msg = response.json()
    if response.status_code != requests.codes.ok:
      self._logger.error('Order submission error: %s', msg)
      if order in self._order_manager:
        self._order_manager.remove_order(order, ignore_error=True)
      self._handle_error_response(response)
      return OrderEvent.ORDER_REJECTED

    exchange_order_id = str(msg['order_id'])
    if exchange_order_id == "-1":
      self._logger.error("Submit order error: %s\n%s", msg, order)
      self._order_manager.remove_order(order, ignore_error=True)
      return OrderEvent.ORDER_ERROR

    prev_order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if prev_order is None:
      self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
    else:
      if prev_order.order_id != order.order_id:
        self._logger.info('order id replaced: %s => %s', prev_order.order_id, order.order_id)
        self._order_manager.remove_order(prev_order)
        self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
        self._order_manager.remove_order(order)
        prev_order.order_id = order.order_id
        self._order_manager.add_order(prev_order)
    return OrderEvent.ORDER_ACCEPTED

  def _handle_cancel_response(self, order, response):
    """
    {
      "client_oid": {},
      "order_id": "1680287676308480",
      "result": true
    }
    {
      "error_message": "You have not uncompleted order at the moment",
      "result": false,
      "error_code": "32004",
      "order_id": "2076276029166593"
    }
    """
    msg = response.json()
    if response.status_code != requests.codes.ok:
      self._logger.error('Cancel order error!')
      order.cancel_sent = False
      self._handle_error_response(response)
      return OrderEvent.CANCEL_ERROR

    if not msg['result']:
      self._logger.error('Cancel order %s failed! Maybe filled already. %s', order.order_id, msg)
      return OrderEvent.CANCEL_ERROR

    exchange_order_id = str(msg['order_id'])
    if exchange_order_id != order.internal.exchange_order_id:
      self._logger.error(
          'Inconsistent exchange_order_id, %s, %s',
          exchange_order_id,
          order.internal.exchange_order_id,
      )
      order.cancel_sent = False
      return OrderEvent.CANCEL_ERROR

    if not self._order_manager.is_dead_order(order):
      self._logger.debug('%s is cancel confirmed.', order.order_id)
      self._order_manager.set_order(order, cancel_confirmed=True)
    else:
      self._logger.info('Order is dead already %s', order)
    return OrderEvent.CANCEL_CONFIRMED

  def _notification_callback(self, data):
    timestamp = get_timestamp()
    channel = data['table']
    try:
      msg_type = self._channel_msg_type_map[channel]
    except KeyError:
      self._logger.error('Unknown message type! %s', data)
      return

    exchange_message = PrivateExchangeMessage(msg_type=msg_type,
                                              data=data,
                                              received_timestamp=timestamp,
                                              request_params=None,
                                              request_timestamp=None)
    self._publisher.write(timestamp, exchange_message)

  @exchange_message_handler(_MsgType.REST_BALANCE)
  def _handle_rest_account_balance(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    self.examine_rest_response_timestamp(exchange_message)
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest account balance response error!')
      self._handle_error_response(response)
      return

    message = response.json()
    self._og_info.update_balance(message)
    self._og_logger.write_balance(self._og_info.balance_as_proto())
    if not self.is_ready():
      self._og_ready.set_balance_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.REST_POSITION)
  def _handle_rest_position(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    self.examine_rest_response_timestamp(exchange_message)
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest position response error!')
      self._handle_error_response(response)
      return

    message = response.json()
    self._og_info.update_position(message)
    self._og_logger.write_position(self._og_info.position_as_proto())
    if not self.is_ready():
      self._og_ready.set_position_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.REST_OPEN_ORDERS)
  def _handle_rest_open_orders(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    self.examine_rest_response_timestamp(exchange_message)
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error!')
      self._handle_error_response(response)
      return

    timestamp = get_timestamp()
    message = response.json()

    exchange_open_order_dict = {}
    for order_info_msg in message['order_info']:
      native_symbol = order_info_msg['instrument_id']
      product = OkexFuturesProduct.FromStrNativeProduct(native_symbol)
      if product not in self._og_info.products:
        continue

      order_info = create_okex_order_from_orderinfo(order_info_msg, timestamp)
      order = self._order_manager.get(exchange_order_id=order_info.internal.exchange_order_id)
      if order:
        order_info.order_id = order.order_id
      else:
        order_info.order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      exchange_open_order_dict[order_info.internal.exchange_order_id] = order_info
      if self.is_ready():
        self._update_order(order_info, timestamp, is_rest=True)

    product = OkexFuturesProduct.FromStrNativeProduct(
        exchange_message.request_params['instrument_id'])
    self._order_manager.copy_from_exchange_open_orders(exchange_open_order_dict, [product])

    if not self.is_ready():
      self._og_ready.set_orders_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.REST_DONE_ORDERS)
  def _handle_rest_done_orders(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    self.examine_rest_response_timestamp(exchange_message)
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error!')
      self._handle_error_response(response)
      return

    timestamp = get_timestamp()
    message = response.json()
    self._num_done_order_query += 1
    for order_msg in message['order_info']:
      order_info = create_okex_order_from_orderinfo(order_msg, timestamp)
      self._update_order(order_info, timestamp, is_rest=True)

  @exchange_message_handler(_MsgType.WS_POSITION)
  def _handle_ws_position(self, exchange_message: PrivateExchangeMessage):
    message = exchange_message.data
    self._og_info.update_ws_position(message)
    self._og_logger.write_position(self._og_info.position_as_proto())

  @exchange_message_handler(_MsgType.WS_ORDER_UPDATE)
  def _handle_ws_trade(self, exchange_message: PrivateExchangeMessage):
    timestamp = get_timestamp()
    message = exchange_message.data
    for order_msg in message['data']:
      order_info = create_okex_order_from_orderinfo(order_msg, timestamp)
      if order_info.internal.status == OkexFuturesOrderStatus.FAILED:
        self._logger.warning('Ws Failed order: %s', message)
      self._update_order(order_info, timestamp, is_rest=False)

  def _update_order(self, order_info, timestamp, is_rest):
    status = order_info.internal.status
    if order_info.product not in self._og_info.products:
      return

    exchange_order_id = order_info.internal.exchange_order_id
    order = self._order_manager.get(exchange_order_id=exchange_order_id)

    last_fill = None
    if order is None:
      # Order submission response has not been received, or it is a foreign order.
      order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      order_info.order_id = order_id
      self._order_manager.add_order(order_info)
      order = order_info
      if order.internal.fill.value.qty_accum_fill > 1e-10:
        last_fill = LastFill(timestamp_last_fill=timestamp,
                             price_last_fill=order.internal.fill.value.price_last_fill,
                             qty_last_fill=order.internal.fill.value.qty_last_fill)
    else:
      last_fill = order.internal.fill.update_by_accum_fill(
          order_info.internal.fill.value.price_avg_fill,
          order_info.internal.fill.value.qty_accum_fill)
      order.internal.fee = order_info.internal.fee

    if is_rest and self._num_done_order_query < 5:
      last_fill = None  # Skip publishing first 5 rest done order queries

    if isinstance(order.tag, dict):
      order_creation_time = order.tag.get('order_creation_time')
      if order_creation_time is not None and self._start_timestamp > order_creation_time:
        last_fill = None

    if last_fill is not None:
      self._og_logger.gen_order_filled_event_and_log(order, timestamp)
      self._fill_stats_calculator.add_fill_by_order(product=order.product,
                                                    filled_qty=abs(last_fill.qty_last_fill),
                                                    order_side=order.order_side,
                                                    order_qty=order.qty)
      order.internal.fill.log_last_fill(order, self._logger)
      self._publish_fill(order.internal.fill.value.price_last_fill,
                         order.internal.fill.value.qty_last_fill,
                         order)
      if is_rest:
        self._logger.info('REST fill is faster! %s', order)

    if status == OkexFuturesOrderStatus.NEW:
      # handled already
      self._logger.debug('order status: %s, order_id=%s', status.name, order.order_id)
    elif status == OkexFuturesOrderStatus.FAILED:
      self._logger.error('Order failed!, %s', order_info)
      self._order_manager.remove_order(order, ignore_error=True)
    elif status == OkexFuturesOrderStatus.CANCELED:
      if not self._order_manager.is_dead_order(order):
        self._order_manager.set_order(order, cancel_confirmed=True)
    elif status == OkexFuturesOrderStatus.PART_FILLED:
      # handled already
      self._logger.debug('order partially filled! %s', order)
    elif status == OkexFuturesOrderStatus.FILLED:
      if order.internal.fill.fully_filled:
        if not self._order_manager.is_dead_order(order):
          self._order_manager.set_order(order, fully_filled=True)
      else:
        self._logger.error('Order fully filled status not consistent! %s', order)
    else:
      self._logger.info('order status: %s, order_id=%s', status.name, order.order_id)


class OkexFuturesOrderGateway(OkexFuturesOrderGatewayBase):
  _exchange = 'Okex'
  _market_type = 'Futures'
  _api_version = 'v3'

  def start(self, queue, ioloop, flow_sub=None):
    api_host = self._og_config.rest_root_url \
        if self._og_config.HasField('rest_root_url') else None
    self._private_client = OkexFuturesNativePrivateClient(key_file=self._config.key_file,
                                                          api_host=api_host,
                                                          use_async=True,
                                                          rest_lib=self._og_config.rest_lib)

    topic_string = 'okex_futureusd_rest_private_1'
    topic_id = topic_map.search_topic_id(topic_string)[0][1]
    self._ioloop = ioloop
    self._publisher = queue.get_writer(topic_id=topic_id)
    self._scheduler = Scheduler(ioloop=ioloop, logger=self._logger)
    self._prepare_private_subscriber()
    self._install_periodic_query()

    # Subscribe to flow for feed.
    self._flow_sub = flow_sub or flow_subscriber.from_queue(queue)
    self._flow_sub.subscribe(topic_string, self.on_private_exchange_msg)

  def _prepare_private_subscriber(self):
    if self._private_subscriber is not None:
      self._private_subscriber.stop()

    def on_ready():
      products = [prod.native_symbol_v3 for prod in self._og_info.products]
      self._private_subscriber.prepare_futures_subscription(products)

    self._private_subscriber = OkexFuturesWsPrivateClient(
        key_file=self._config.key_file,
        ioloop=self._ioloop,
        notification_callback=self._notification_callback,
        close_callback=functools.partial(self._prepare_private_subscriber),
        on_ready=on_ready,
    )
    self._private_subscriber.start()
